package com.nice.rm.core;

import java.io.DataInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.analysis.standard.StandardAnalyzer;
import org.apache.lucene.document.Document;
import org.apache.lucene.index.DirectoryReader;
import org.apache.lucene.index.IndexReader;
import org.apache.lucene.index.IndexWriter;
import org.apache.lucene.index.IndexWriterConfig;
import org.apache.lucene.queryparser.classic.QueryParser;
import org.apache.lucene.search.IndexSearcher;
import org.apache.lucene.search.Query;
import org.apache.lucene.search.ScoreDoc;
import org.apache.lucene.search.TopDocs;
import org.apache.lucene.store.Directory;
import org.apache.lucene.store.FSDirectory;
import org.apache.lucene.store.RAMDirectory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.nice.rm.event.EvenType;
import com.nice.rm.event.Event;
import com.nice.rm.event.EventDispatcher;
import com.nice.rm.event.EventHandler;
import com.nice.rm.util.FileUtil;

public class RealTimeIndexManager implements EventHandler {

    private static final Logger log = LoggerFactory.getLogger(RealTimeIndexManager.class);
    public static final String INDEX_DIR_SUFFIX = "index_";
    public static final String INDEX_LOG_SUFFIX = "log_";

    private File path;
    private AtomicLong count;
    private Analyzer analyzer;
    private int memoryDataSize;
    private IndexWriter radWriter;
    private IndexReader radReader;
    private IndexReader fsReader;
    private AtomicLong writeIndex;
    private LogAppendWriter logWriter;
    private EventDispatcher eDispatchCenter;
    private ExpireIndexFileGCService gcService;

    public RealTimeIndexManager(String indexPath, int memoryDataSize) {
        try {
            this.path = new File(indexPath);
            this.count = new AtomicLong();
            this.memoryDataSize = memoryDataSize;
            this.analyzer = new StandardAnalyzer();
            this.writeIndex = new AtomicLong();
            this.gcService = new ExpireIndexFileGCService(path);
            this.eDispatchCenter = EventDispatcher.getInstance();

            RAMDirectory ramDirectory = new RAMDirectory();
            IndexWriterConfig cfg = new IndexWriterConfig(analyzer);
            this.radWriter = new IndexWriter(ramDirectory, cfg);
            this.radReader = DirectoryReader.open(radWriter, true);

            if (!this.path.exists())
                this.path.mkdirs();

            if (memoryDataSize == 0)
                this.memoryDataSize = 100000;

            this.openFileSystemIndexIfExist();
            this.gcService.start();

            File LogFile = new File(path, "log_" + writeIndex.get());
            this.logWriter = new LogAppendWriter(LogFile);

            this.eDispatchCenter.regist(this);
            this.eDispatchCenter.regist(new PersistentService());
            this.eDispatchCenter.startEventLoop();

            log.info(FileUtil.strCat("init success", "memoIndex=", radReader, "fIndex=", fsReader));
        } catch (IOException e) {
            log.error("init  error", e);
            throw new RuntimeException(e);
        }
    }

    public synchronized <T> int query(QueryParameter<T> param) {
        try {
            int total = queryInMemoryIndex(param);
            if (param.data.size() < param.pageSize) {
                total += queryInFileIndex(param);
            }
            return total;
        } catch (Exception e) {
            log.error("query  error", e);
            throw new RuntimeException(e);
        }
    }

    public synchronized boolean processDoc(Document doc, String line) {
        try {
            logWriter.appendWrite(line);
            radWriter.addDocument(doc);
            writeToFileIfReachThreshold();
            return true;
        } catch (IOException e) {
            close();
            log.error("add index  error", e);
            throw new RuntimeException(e);
        }

    }

    public synchronized void close() {
        try {
            this.writeIndexToFile(true);
            this.gcService.shutdown();
            this.eDispatchCenter.shutdown();
        } catch (IOException e) {
            log.error("close index  error", e);
        }

    }

    /**
     * 在指定的index里查找
     * 
     * @param param
     * @param searcher
     * @return
     * @throws Exception
     */
    private <T> int queryInSpecIndex(QueryParameter<T> param, IndexSearcher searcher) throws Exception {
        List<T> out = param.data;
        int pageSize = param.pageSize;
        int start = param.page * pageSize;
        int realSize = (param.page + 1) * pageSize;
        Query query = new QueryParser(param.field, analyzer).parse(param.key);
        TopDocs docs = searcher.search(query, realSize);
        ScoreDoc[] socreDocs = docs.scoreDocs;
        for (int i = start; i < socreDocs.length; i++) {
            if (out.size() == pageSize)
                break;
            Document doc = searcher.doc(socreDocs[i].doc);
            out.add(param.callback.docToObj(doc));
        }
        return docs.totalHits;
    }

    @SuppressWarnings("deprecation")
    private void openFileSystemIndexIfExist() throws IOException {
        File lastFileName = new File(path, PersistentService.LAST_INDEX_DIR);
        if (lastFileName.exists()) {
            SimpleDateFormat fmt = new SimpleDateFormat("yy-M-d H:m:s");
            FileInputStream fis = new FileInputStream(lastFileName);
            DataInputStream dis = new DataInputStream(fis);
            long lastFileIndex = dis.readLong();
            long lastSaveTime = dis.readLong();
            String lastIndexDirName = dis.readLine();
            if (lastIndexDirName != null && new File(lastIndexDirName).exists()) {
                try {
                    if (lastFileIndex > 0)
                        writeIndex.set(lastFileIndex + 1);
                    Path lastIndex = Paths.get(lastIndexDirName);
                    FSDirectory open = FSDirectory.open(lastIndex);
                    fsReader = DirectoryReader.open(open);
                    String time = fmt.format(new Date(lastSaveTime));
                    log.info(String.format("reload index from:%s time:%s", lastIndexDirName, time));
                } catch (Exception e) {
                    log.error("reload index error:" + lastIndexDirName, e);
                }
            }
            FileUtil.close(fis, dis);
        }
    }

    private void writeToFileIfReachThreshold() throws IOException {
        if (count.getAndIncrement() >= memoryDataSize) {
            writeIndexToFile(false);
            count.set(0);
        }
    }

    private void writeIndexToFile(boolean isClose) throws IOException {
        // 如果关闭时没数据,就不需要持久化
        if (count.get() == 0)
            return;

        long curIndex = writeIndex.get();
        if (isClose == false) {
            writeIndex.incrementAndGet();
        } else {
            if (count.get() >= memoryDataSize) {
                // 如果关闭时数据达到memoryDataSize才自增
                writeIndex.incrementAndGet();
            }
        }
        IndexWriter oldIdWriter = radWriter;
        LogAppendWriter oldWriter = logWriter;
        File oldLogFile = oldWriter.getFile();

        // 关闭radWriter很耗时,在异步线程处理
        FileUtil.close(radReader, logWriter);
        File oldIndexDir = FileUtil.buildDir(path, INDEX_DIR_SUFFIX, curIndex);

        Event event = new Event();
        event.type = EvenType.PERSIST;
        event.message = new PersistentEvent(oldIdWriter, oldLogFile, analyzer, oldIndexDir, curIndex);
        eDispatchCenter.sendEvent(event);

        File newLog = FileUtil.buildDir(path, INDEX_LOG_SUFFIX, writeIndex.get());
        IndexWriterConfig newCfg = new IndexWriterConfig(analyzer);
        RAMDirectory ramDirectory = new RAMDirectory();

        radWriter = new IndexWriter(ramDirectory, newCfg);
        logWriter = new LogAppendWriter(newLog);
        radReader = DirectoryReader.open(radWriter, true);

        // index持久化后在更新fsReader

    }

    private <T> int queryInFileIndex(QueryParameter<T> param) throws Exception {
        if (fsReader == null)
            return 0;
        IndexSearcher fsSearcher = new IndexSearcher(fsReader);
        return queryInSpecIndex(param, fsSearcher);
    }

    private <T> int queryInMemoryIndex(QueryParameter<T> param) throws Exception {
        DirectoryReader tmp = (DirectoryReader) radReader;
        if (!tmp.isCurrent()) {
            DirectoryReader newReader = DirectoryReader.openIfChanged(tmp, radWriter, true);
            if (newReader != null) {
                radReader.close();
                radReader = newReader;
            }
        }
        IndexSearcher searcher = new IndexSearcher(radReader);
        return queryInSpecIndex(param, searcher);
    }

    @Override
    public void handle(Event event) {
        try {
            Directory dir = (Directory) event.message;
            IndexReader newFsReader = DirectoryReader.open(dir);
            synchronized (this) {
                if (fsReader != null)
                    FileUtil.close(fsReader);
                fsReader = newFsReader;
            }
        } catch (Exception e) {
            log.error("update fileIndex  error:" + event.message, e);
        }
    }

    @Override
    public boolean match(EvenType type) {
        return type == EvenType.UPDATE_INDEX;
    }

}
